package com.hex.es.bulk;

import com.alibaba.fastjson.JSON;
import com.hex.es.entity.EsInfoDomain;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 信息批量存储入es的中间缓存类
 *
 * @author hex
 * @date 2019/11/01
 */
@Slf4j
public class EsInfoStorageCache {

    /**异常数据最大允许获取次数*/
    private final static int ERROR_MAX_COUNT = 3;

    /**计数器*/
    private volatile static AtomicInteger counter = new AtomicInteger(0);

    /**存放正常数据的队列*/
    private final static ConcurrentLinkedQueue<EsInfoDomain> QUEUE = new ConcurrentLinkedQueue<>();

    /**存放处理失败的异常队列*/
    private final static ConcurrentLinkedQueue<EsInfoDomain> ERROR_QUEUE = new ConcurrentLinkedQueue<>();

    public static void add(EsInfoDomain esInfoDomain){
        QUEUE.offer(esInfoDomain);
    }

    public static void addAllOfErrorQueue(List<EsInfoDomain> esInfoDomains){
        if (CollectionUtils.isNotEmpty(esInfoDomains)) {
            ERROR_QUEUE.addAll(esInfoDomains);
        }
    }

    public static List<EsInfoDomain> get(){
        return get(1);
    }

    /**
     * 获取数据
     * 1、优先从异常队列获取
     * 2、  如果异常队列为空，再从正常队列获取
     *      如果异常队列已经处理三次，也直接从正常队列获取
     *
     * @return
     */
    public static List<EsInfoDomain> get(int count){
        if (ERROR_QUEUE.size() > 0 && counter.get() < ERROR_MAX_COUNT) {
            counter.incrementAndGet();
            List<EsInfoDomain> esInfoDomains = get(count, ERROR_QUEUE);
            log.warn("EsOperationLogicRunnable get data from error_queue. warn!!!!data:{}", JSON.toJSONString(esInfoDomains));
            return esInfoDomains;
        } else {
            counter.set(0);
            return get(count,QUEUE);
        }
    }

    private static List<EsInfoDomain> get(int count,ConcurrentLinkedQueue<EsInfoDomain> queue){
        List<EsInfoDomain> data = new ArrayList();
        if (queue != null) {
            for (int i = 1; i <= count; i++) {
                EsInfoDomain poll = queue.poll();
                if (poll == null) {
                    return data;
                }
                data.add(poll);
            }
        }
        return data;
    }

    public static int size(){
        return QUEUE.size();
    }
}
